package cn.xuqiudong.rpc.registry.zookeeper;

import cn.xuqiudong.rpc.common.model.XqdUrl;
import cn.xuqiudong.rpc.common.util.XqdUrlUtils;
import org.apache.commons.collections4.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import cn.xuqiudong.rpc.cluster.exception.XqdCommonException;
import cn.xuqiudong.rpc.registry.NotifyListener;
import cn.xuqiudong.rpc.registry.Registry;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * 描述:基于zookeeper的注册中心
 * @author Vic.xu
 * @date 2022-02-15 11:34
 */
public class ZookeeperRegistry implements Registry {

    private static Logger logger = LoggerFactory.getLogger(ZookeeperRegistry.class);

    private XqdZookeeperClient client;

    private String registryAddress;

    /**
     * 缓存需要通知的监听器
     */
    private final ConcurrentMap<XqdUrl, ConcurrentMap<NotifyListener, ChildListener>> zkListeners = new ConcurrentHashMap<>();

    /**
     * 缓存需要通知的url
     */
    private final ConcurrentMap<XqdUrl, Map<String, List<XqdUrl>>> notified = new ConcurrentHashMap<>();

    public ZookeeperRegistry(String address, XqdZookeeperClient client) {
        this.registryAddress = address;
        this.client = client;

    }

    /**
     * 注册， 创建一个临时节点
     * @param url
     */
    @Override
    public void register(XqdUrl url) {
        logger.info("向注册中心注册{}接口 {}:{}", url.getInterfaceName(), url.getAddress(), url.getPort());
        try{
            client.create(XqdUrlUtils.toPath(url),true);
        }catch (Exception e) {
            e.printStackTrace();
            throw new XqdCommonException("注册url:" + url + "到zookeeper " +registryAddress +" 失败", e);
        }
    }

    /**
     * 取消注册
     * @param url
     */
    @Override
    public void unregister(XqdUrl url) {
        try{
            client.delete(XqdUrlUtils.toPath(url));
        }catch (Exception e) {
            throw new XqdCommonException("取消注册url:" + url + " 从zookeeper " +registryAddress +" 失败", e);
        }
    }

    @Override
    public void subscribe(XqdUrl url, NotifyListener listener) {
        logger.info("向注册中心订阅服务{}地址", url.getInterfaceName());
        String parent = XqdUrlUtils.findParent(url);
        //把当前url放入到zkListeners中
        ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.computeIfAbsent(url, k -> new ConcurrentHashMap<>());
        //ChildListener  之前不存在，则创建监听，监听的回调为this#notify,
        ChildListener zkListener = listeners.computeIfAbsent(listener, k -> (parentPath, currentChilds) ->
                ZookeeperRegistry.this.notify(url, k, XqdUrlUtils.toUrls(currentChilds)));
        //创建临时的父节点
        client.create(parent, false);
        // 向zookeeper的父节点发起订阅
        List<String> children = client.addChildListener(parent, zkListener);
    }

    /**
     * 取消订阅
     * @param url
     * @param listener
     */
    @Override
    public void unsubscribe(XqdUrl url, NotifyListener listener) {
        ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.get(url);
        if (listeners != null) {
            ChildListener zkListener = listeners.get(listener);
            if (zkListener != null) {
                String parent = XqdUrlUtils.findParent(url);
                client.removeChildListener(parent, zkListener);
            }
        }
    }

    @Override
    public List<XqdUrl> lookup(XqdUrl url) {
        if (url ==null) {
            throw new XqdCommonException("Query url cannot be null");
        }
        String parent = XqdUrlUtils.findParent(url);
        List<String> children = client.getChildren(parent);
        if (CollectionUtils.isEmpty(children)) {
            return null;
        }
        List<XqdUrl> list = XqdUrlUtils.toUrls(children);
        return list;
    }

    /**
     * 通知服务变化 回调NotifyListener
      */
    protected void notify(XqdUrl url, NotifyListener listener, List<XqdUrl> urls) {
        if (url == null) {
            throw  new XqdCommonException(" url cannot be null");
        }

        if (listener == null) {
            throw  new XqdCommonException(" listener cannot be null");
        }
        //先从缓存中获取 然后覆盖缓存
        Map<String, List<XqdUrl>> needNotified = notified.computeIfAbsent(url, u -> new ConcurrentHashMap<>());
        needNotified.put(XqdUrlUtils.findParent(url), urls);
        listener.notify(urls);
    }

}
